#include <poll.h>
#include <stdint.h>
#include <sys/un.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <errno.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/prctl.h>


#include "list.h"
#include "pb.h"
#include "pb_encode.h"
#include "pb_decode.h"

#include "somb.h"
#include "somb_manager.h"
#include "somb_daemon.pb.h"
#include "somb_log.h"
#include "somb_work.h"
#include "somb_lock.h"

pthread_mutex_t SubsNsMutexLock = PTHREAD_MUTEX_INITIALIZER;
ST_SOMB_LIB_SERVICE_LIST *SubNsServieList= NULL;


static somb_spinlock_t SubNsListLock;
struct somb_mngr_sub_ns *SubNsList=NULL;

static struct somb_local_sub_ns *pstLocalSubNs = NULL;

int32_t i32SOMB_Msg_Send_2_Daemon(uint32_t MsgId, void *msgreq,const pb_msgdesc_t * reqfields,uint32_t reqfield_size);
int32_t i32SOMB_Manager_Poll(uint32_t MsgId,uint32_t timeout_ms);


int32_t somb_manager_connect_remote_subns(struct somb_mngr_sub_ns *pstSubNs)
{
    //close sk first
    if(pstSubNs->skfd != 0)
        close(pstSubNs->skfd);

    pstSubNs->skfd = socket(AF_UNIX,SOCK_STREAM|SOCK_CLOEXEC,0);
    struct sockaddr_un dest_addr_un;
    memset(&dest_addr_un,0,sizeof(struct sockaddr_un));
    dest_addr_un.sun_family =  AF_UNIX;
    snprintf(dest_addr_un.sun_path,sizeof(dest_addr_un.sun_path),"%s",pstSubNs->sub_ns_addr);
    //sprintf(dest_addr_un.sun_path,"%s",pstSubNs->sub_ns_addr);

    struct sockaddr_un src_addr_un;
    memset(&src_addr_un,0,sizeof(struct sockaddr_un));
    src_addr_un.sun_family =  AF_UNIX;
    sprintf(src_addr_un.sun_path,SOMB_SUB_NS_ADDR_FMT2,pstLocalSubNs->sub_ns_id,pstSubNs->sub_ns_id);
    unlink(src_addr_un.sun_path);
    if(bind(pstSubNs->skfd,(struct sockaddr*)&src_addr_un,SUN_LEN(&src_addr_un)) < 0)
    {
        SOMB_LOG_ERR("somb mng:(%s) pid(%d) subns_id(%d) bind addr for subns_id(%d) with SrcAddr(%s) failed",pstLocalSubNs->localexename, pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,pstSubNs->sub_ns_id,src_addr_un.sun_path);
    }
    else
    {
        SOMB_LOG_DEBUG("somb mng:(%s) pid(%d) subns_id(%d) bind addr for subns_id(%d) with SrcAddr(%s) success",pstLocalSubNs->localexename, pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,pstSubNs->sub_ns_id,src_addr_un.sun_path);

    }

    if(connect(pstSubNs->skfd,(struct sockaddr*)&dest_addr_un,SUN_LEN(&dest_addr_un)) < 0)
    {
        pstSubNs->SubNsStatus = SOMB_SUBNS_STATUS_OFFLINE;
        SOMB_LOG_ERR("somb mng:(%s) pid(%d) subns_id(%d) connect to new subns_id(%d) (%s) failed,errno=%s",pstLocalSubNs->localexename, pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,pstSubNs->sub_ns_id,pstSubNs->sub_ns_addr,strerror(errno));
        return SOMB_FAIL;
    }
    else
    {
        pstSubNs->SubNsStatus = SOMB_SUBNS_STATUS_ONLINE;
        SOMB_LOG_INFO("somb mng:(%s) pid(%d) subns_id(%d) connect to new subns_id(%d) (%s) success,sk=%d",pstLocalSubNs->localexename, pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,pstSubNs->sub_ns_id,pstSubNs->sub_ns_addr,pstSubNs->skfd);

        return SOMB_OK;
    }
}

struct somb_mngr_sub_ns *pstSOMB_GetSubNsBySkfd2(uint32_t skfd2)
{
    struct somb_mngr_sub_ns *pstSubNs=NULL,*tmp = NULL;

    somb_spinlock_acquire(&SubNsListLock);
    HASH_ITER(hh, SubNsList, pstSubNs, tmp)
    {
        if((NULL != pstSubNs)&&(pstSubNs->skfd2 == skfd2))
        {
            somb_spinlock_release(&SubNsListLock);
            return pstSubNs;
        }
    }

    somb_spinlock_release(&SubNsListLock);
    return NULL;
}
struct somb_mngr_sub_ns *pstSOMB_Manager_Find_SubNs(uint32_t sub_ns_id)
{
    struct somb_mngr_sub_ns *tmp = NULL;
    somb_spinlock_acquire(&SubNsListLock);
    HASH_FIND_INT(SubNsList,&(sub_ns_id),tmp) ;
    somb_spinlock_release(&SubNsListLock);
    return tmp;
}
struct somb_mngr_sub_ns *pstSOMB_GetSubNsById(uint32_t sub_ns_id)
{
    int ret;
    SOMB_LOG_DEBUG("somb mng:(%s) pid(%d) sub ns id(%d) get sub ns(%d) start",pstLocalSubNs->localexename,
                    pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,sub_ns_id);

    struct somb_mngr_sub_ns *tmp = NULL;

    somb_spinlock_acquire(&SubNsListLock);
    HASH_FIND_INT(SubNsList,&(sub_ns_id),tmp) ;
    somb_spinlock_release(&SubNsListLock);

    if(NULL == tmp)
    {
        //get service from daemon
        somb_msg_get_sub_ns_req msgreq = somb_msg_get_sub_ns_req_init_zero;
        msgreq.sub_ns_id = pstLocalSubNs->sub_ns_id;
        msgreq.pid = pstLocalSubNs->localpid;

        msgreq.has_subns = true;
        msgreq.subns.sub_ns_id = sub_ns_id;

        pthread_mutex_lock(&(pstLocalSubNs->sk_lock));
        ret = i32SOMB_Msg_Send_2_Daemon(somb_daemon_msgid_get_sub_ns,\
            &msgreq,somb_msg_get_sub_ns_req_fields,somb_msg_get_sub_ns_req_size);
        if(SOMB_OK != ret)
        {
            SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) get sub ns(%d) failed!",pstLocalSubNs->localexename,
                    pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,sub_ns_id);
            pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
            goto fail1;
        }

        ret = i32SOMB_Manager_Poll(somb_daemon_msgid_get_sub_ns,500);
        if(ret != SOMB_OK)
        {
            SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) get sub ns(%d) wait response failed!",pstLocalSubNs->localexename, pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,sub_ns_id);
            pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
            goto fail1;
        }

        somb_msg_get_sub_ns_resp *msgresp = (somb_msg_get_sub_ns_resp *)pstLocalSubNs->cond_MsgData;

        tmp = (struct somb_mngr_sub_ns*)malloc(sizeof *tmp);
        if(NULL == tmp)
        {
            SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) get sub ns(%d) malloc failed!",pstLocalSubNs->localexename, pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,sub_ns_id);
            free(pstLocalSubNs->cond_MsgData);
            pstLocalSubNs->cond_MsgData=NULL;
            pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
            goto fail1;
        }

        memset(tmp,0,sizeof(struct somb_mngr_sub_ns));
        tmp->sub_ns_id = msgresp->subns.sub_ns_id;
        tmp->resp_data_head = NULL;

        pthread_condattr_t astCondAttr;
        pthread_condattr_setclock(&astCondAttr , CLOCK_MONOTONIC);
        pthread_cond_init(&(tmp->cond) , &astCondAttr);

        pthread_mutex_init(&tmp->mutex_lock,NULL);

        sprintf(tmp->sub_ns_addr,"%s",msgresp->subns.sub_ns_addr);
        sprintf(tmp->sub_ns_name,"%s",msgresp->subns.sub_ns_name);
        SOMB_LOG_INFO("somb mng:(%s) pid(%d) subns_id(%d) add new subns_id(%d) addr(%s)!",pstLocalSubNs->localexename, pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,tmp->sub_ns_id,tmp->sub_ns_addr);

        pthread_condattr_destroy(&astCondAttr);

        free(pstLocalSubNs->cond_MsgData);
        pstLocalSubNs->cond_MsgData=NULL;

        somb_manager_connect_remote_subns(tmp);
        somb_spinlock_acquire(&SubNsListLock);
        HASH_ADD_INT(SubNsList, sub_ns_id, tmp);  /* id is the key field */
        somb_spinlock_release(&SubNsListLock);
        pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
    }
    SOMB_LOG_DEBUG("somb mng:(%s) pid(%d) sub ns id(%d) get sub ns(%d) finish",pstLocalSubNs->localexename,
                    pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,sub_ns_id);
    return tmp;
fail1:
    return NULL;
}

struct somb_local_sub_ns *pstSOMB_GetLocalSubNs(void)
{
    //printf("pstSOMB_GetLocalSubNs pstLocalSubNs=%p\n",pstLocalSubNs);
    return pstLocalSubNs;
}

int32_t i32SOMB_Manager_Poll(uint32_t MsgId,uint32_t timeout_ms)
{
    int32_t ret;
    struct timespec abstime;
    clock_gettime(CLOCK_MONOTONIC, &abstime);
    uint64_t tmp_nsec=  (uint64_t)(abstime.tv_nsec) + timeout_ms*1E6;
    abstime.tv_sec += tmp_nsec/1000000000;
    abstime.tv_nsec = tmp_nsec%1000000000;

    if(NULL != pstLocalSubNs->cond_MsgData)
    {
        SOMB_LOG_ERR("somb msg:subns(%s) pid(%d) timedwait for MsgId(%d)'s response start,but cond_MsgData is not NUL;(%p)",pstLocalSubNs->localexename,pstLocalSubNs->localpid,MsgId,pstLocalSubNs->cond_MsgData);
    }

    do
    {
       ret =  pthread_cond_timedwait(&(pstLocalSubNs->sk_cond),&(pstLocalSubNs->sk_lock),&abstime);
       if(ret != 0)
       {
           if((EINVAL == ret) || (ETIMEDOUT ==  ret))
           {
               ret =  SOMB_FAIL;
                SOMB_LOG_ERR("somb msg:subns(%s) pid(%d) timedwait for MsgId(%d)'s response failed",pstLocalSubNs->localexename,pstLocalSubNs->localpid,MsgId);
               break;
           }

           if ((EINTR == errno) || (EAGAIN == errno))
           {

                SOMB_LOG_ERR("somb msg:subns(%s) pid(%d) timedwait for MsgId(%d)'s response failed,errno=%s,try again",pstLocalSubNs->localexename,pstLocalSubNs->localpid,MsgId,strerror(errno));
               continue;
           }
           else
           {
               ret =  SOMB_FAIL;
                SOMB_LOG_ERR("somb msg:subns(%s) pid(%d) timedwait for MsgId(%d)'s response failed,errno=%s",pstLocalSubNs->localexename,pstLocalSubNs->localpid,MsgId,strerror(errno));
               break;
           }

       }
       else
       {
           if(MsgId == pstLocalSubNs->cond_MsgId) 
           {
                SOMB_LOG_DEBUG("somb msg:subns(%s) pid(%d) timedwait for MsgId(%d)'s response success",pstLocalSubNs->localexename,pstLocalSubNs->localpid,MsgId);
               ret =  SOMB_OK;
               break;
           }
           else
           {
                SOMB_LOG_ERR("somb msg:subns(%s) pid(%d) timedwait for MsgId(%d)'s response failed,but expected=%d",pstLocalSubNs->localexename,pstLocalSubNs->localpid,MsgId,pstLocalSubNs->cond_MsgId);
                usleep(100*1000);
           }
       }
    }while(1);

    return ret;

}
int32_t i32SOMB_Manager_SubNs_Init(uint32_t *pSubNsId)
{
    int32_t ret = SOMB_OK;
    struct sockaddr_un tmp_un_addr;
    if(NULL == pstLocalSubNs)
    {

        uint32_t pid = getpid();
        pstLocalSubNs = malloc(sizeof(struct somb_local_sub_ns));
        if(NULL == pstLocalSubNs)
        {
            SOMB_LOG_ERR("somb mng: pid(%d) malloc for sub ns failed",pid);
            return SOMB_FAIL;
        }
        memset(pstLocalSubNs,0,sizeof(struct somb_local_sub_ns));


        pstLocalSubNs->localpid=  pid;
        pstLocalSubNs->sub_ns_id = 0;

        char strProcessPath[1024] = {0};
        char *strProcessName = NULL;
        //if(readlink("/proc/self/exe", strProcessPath,1024) <=0)
        if(0 != prctl(PR_GET_NAME,strProcessPath,0,0,0))
        {
            SOMB_LOG_ERR("somb mng: get process name failed");
            sprintf(strProcessPath,"pid%d",pstLocalSubNs->localpid);
            strProcessName = strProcessPath;
        }
        else
        {
            //strProcessName = strrchr(strProcessPath, '/');
            //strProcessName++;

            strProcessName = strProcessPath;
            strProcessName = strProcessPath;
            int len = strlen(strProcessName);
            int i, j = 0;

            for (i = 0; i < len; i++) {
                // 如果当前字符不是'-'，则将其移到新的字符串中
                if (strProcessName[i] != '-') {
                    strProcessName[j++] = strProcessName[i];
                }
            }
            // 添加字符串结束标志
            strProcessName[j] = '\0';

        }
        sprintf(pstLocalSubNs->localexename,"%s",strProcessName);

        pstLocalSubNs->sk = socket(AF_UNIX,SOCK_STREAM|SOCK_CLOEXEC,0);
        sprintf(pstLocalSubNs->sub_ns_addr,SOMB_SUB_NS_ADDR_FMT0,pstLocalSubNs->localpid,pstLocalSubNs->localexename);

        memset(&(tmp_un_addr),0,sizeof(tmp_un_addr));
        tmp_un_addr.sun_family = AF_UNIX;
        snprintf(tmp_un_addr.sun_path,sizeof(tmp_un_addr.sun_path),"%s",pstLocalSubNs->sub_ns_addr);
        unlink(pstLocalSubNs->sub_ns_addr);
        if(bind(pstLocalSubNs->sk,(struct sockaddr *)&tmp_un_addr,SUN_LEN(&tmp_un_addr))<0)
        {
            SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) bind addr(%s) failed(%s)",pstLocalSubNs->localexename,\
                    pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,pstLocalSubNs->sub_ns_addr,strerror(errno));
            goto fail1;
        }

        struct sockaddr_un ServerAddr_un;
        memset(&(ServerAddr_un),0,sizeof(ServerAddr_un));
        ServerAddr_un.sun_family = AF_UNIX;
        snprintf(ServerAddr_un.sun_path,sizeof(ServerAddr_un.sun_path),"%s",SOMB_DAEMON_ADDR);
        if(connect(pstLocalSubNs->sk,(struct sockaddr*)&ServerAddr_un,(socklen_t)SUN_LEN(&ServerAddr_un)) < 0)
        {
            SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) addr(%s) connect somb_daemon  failed(%s)",pstLocalSubNs->localexename, pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,pstLocalSubNs->sub_ns_addr,strerror(errno));
            goto fail1;
        }

        pthread_condattr_t astCondAttr;
        pthread_condattr_setclock(&astCondAttr , CLOCK_MONOTONIC);
        pthread_cond_init(&(pstLocalSubNs->sk_cond) , &astCondAttr);
        pthread_mutex_init(&(pstLocalSubNs->sk_lock),NULL);
        pthread_condattr_destroy(&astCondAttr);
        pstLocalSubNs->cond_MsgId = 0;
        pstLocalSubNs->cond_MsgData = NULL;
        pthread_mutex_init(&(pstLocalSubNs->SubNsListLock),NULL);

        somb_spinlock_init(&SubNsListLock);
    }
    else
    {
        SOMB_LOG_WARN("somb mng:(%s) pid(%d) sub ns id(%d) addr(%s) already Initialized",pstLocalSubNs->localexename,\
                    pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,pstLocalSubNs->sub_ns_addr);
        return SOMB_OK;
    }



    somb_sub_ns_reg_req msgreq = somb_sub_ns_reg_req_init_zero;
    msgreq.pid = pstLocalSubNs->localpid;
    snprintf(msgreq.sub_ns_name,sizeof(msgreq.sub_ns_name),"%s",pstLocalSubNs->localexename);
    uint8_t MsgId = somb_daemon_msgid_register_sub_ns;

    pthread_mutex_lock(&(pstLocalSubNs->sk_lock));
    //SOMB_LOG_DEBUG("somb mng: reqfields=%p reqfield_size=%d",somb_sub_ns_reg_req_fields,somb_sub_ns_reg_req_size);
    ret = i32SOMB_Msg_Send_2_Daemon(somb_daemon_msgid_register_sub_ns,\
            &msgreq,somb_sub_ns_reg_req_fields,somb_sub_ns_reg_req_size);
    if(SOMB_OK != ret)
    {
        SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) reg to daemon failed!",pstLocalSubNs->localexename,
                pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id);
        pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
        goto fail2;
    }

    //somb_sub_ns_reg_resp *msgresp = (somb_sub_ns_reg_resp *)pstLocalSubNs->cond_MsgData;
    somb_sub_ns_reg_resp msgresp = somb_sub_ns_reg_resp_init_zero;
    uint8_t buffer1[SOMB_DAEMON_MSG_HEADER_SIZE+somb_sub_ns_reg_resp_size ];
    memset(buffer1,0,sizeof(buffer1));

    uint32_t recvlen = recvfrom(pstLocalSubNs->sk,buffer1,sizeof(buffer1),0,NULL,0);
    if(recvlen == -1)
    {
        SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) reg to daemon recv response len failed.errno=%s!",pstLocalSubNs->localexename, pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,strerror(errno));
        pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
        goto fail2;
    }

    uint32_t msgid = *(uint32_t *)buffer1;
    SOMB_LOG_DEBUG("somb mng:(%s) pid(%d) sub ns id(%d) send MsgId(%d) to daemon, recv response msgid=%d len=%d",pstLocalSubNs->localexename,pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,MsgId,msgid,recvlen);

    pb_istream_t istream = pb_istream_from_buffer(&(buffer1[SOMB_DAEMON_MSG_HEADER_SIZE]),recvlen - SOMB_DAEMON_MSG_HEADER_SIZE);
    if(msgid != MsgId)
    {
        SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) send MsgId(%d) to daemon, but recv response MsgId(%d)! failed!",pstLocalSubNs->localexename,pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,MsgId,msgid);
        ret = SOMB_MSG_ID_ERR;
        pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
        goto fail2;
    }

    if(false == pb_decode(&istream,somb_sub_ns_reg_resp_fields , &msgresp))
    {
        SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) send MsgId(%d) to daemon,but decode failed!(%s)",pstLocalSubNs->localexename,pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,MsgId,PB_GET_ERROR(&istream));
        ret = SOMB_MSG_DECODE_ERR;
        pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
        goto fail2;
    }

    pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));

    pstLocalSubNs->sub_ns_id = msgresp.sub_ns_id;
    if(msgresp.sub_ns_id == 0)
    {
        SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) send MsgId(%d) to daemon,but daemon return subns is 0",pstLocalSubNs->localexename,pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,MsgId);
        ret = SOMB_REG_FAILED;
        goto fail2;
    }

    sprintf(pstLocalSubNs->sub_ns_addr1,SOMB_SUB_NS_ADDR_FMT1,pstLocalSubNs->sub_ns_id);

    pstLocalSubNs->sk1 = socket(AF_UNIX,SOCK_STREAM|SOCK_CLOEXEC,0);
    memset(&(tmp_un_addr),0,sizeof(tmp_un_addr));
    tmp_un_addr.sun_family = AF_UNIX;
    snprintf(tmp_un_addr.sun_path,sizeof(tmp_un_addr.sun_path),"%s",pstLocalSubNs->sub_ns_addr1);
    unlink(pstLocalSubNs->sub_ns_addr1);
    if(bind(pstLocalSubNs->sk1,(struct sockaddr *)&tmp_un_addr,SUN_LEN(&tmp_un_addr))<0)
    {
        SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) bind addr(%s) failed",pstLocalSubNs->localexename,\
                pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,pstLocalSubNs->sub_ns_addr1);
        goto fail2;
    }

    if(listen(pstLocalSubNs->sk1,10) < 0)
    {
        SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) listen addr(%s) failed",pstLocalSubNs->localexename,\
                pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,pstLocalSubNs->sub_ns_addr1);
        goto fail2;
    }

    if(NULL == pstSOMB_Manager_SaveSubNs(pstLocalSubNs->sub_ns_id,pstLocalSubNs->sub_ns_addr1,pstLocalSubNs->localexename))
    {
        SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) save local sub ns to manager failed ",pstLocalSubNs->localexename,pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id);
        goto fail2;
    }

    ret = SOMB_Worker_Init();
    if(SOMB_OK != ret)
    {
        SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) reg to daemon failed",pstLocalSubNs->localexename,\
            pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id);
        goto fail2;
    }

    SOMB_LOG_INFO("somb mng:(%s) pid(%d) sub ns id(%d) reg to daemon success",pstLocalSubNs->localexename,\
            pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id);
    *pSubNsId = pstLocalSubNs->sub_ns_id;
    return ret;

fail2:
    free(pstLocalSubNs);
    pstLocalSubNs = NULL;
fail1:
    return SOMB_FAIL;
}

//struct somb_mngr_service_item *pstSOMB_Manager_SaveService(uint32_t ServiceId,uint8_t ServiceType,void *Priv,uint32_t sub_ns_id,char sub_ns_addr[],char sub_ns_name[])
struct somb_mngr_service_item *pstSOMB_Manager_SaveService(uint32_t ServiceId,uint8_t ServiceType,void *Priv,struct somb_mngr_sub_ns *pstSubNs)
{
    ST_SOMB_LIB_SERVICE_LIST *s=NULL;
    HASH_FIND_INT(SubNsServieList, &ServiceId, s);  /* id already in the hash? */
    if (s == NULL) {
        s = (ST_SOMB_LIB_SERVICE_LIST*)malloc(sizeof *s);
        if(NULL == s)
        {
            SOMB_LOG_ERR("somb mng:(%s) sub ns(%d) add new ServiceId(%08x) ServiceType(%d) malloc failed",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,ServiceId,ServiceType);
            return NULL;
        }
        memset(s,0,sizeof(ST_SOMB_LIB_SERVICE_LIST));

        s->ServiceId = ServiceId;
        s->ServiceType = ServiceType;
        s->ServicePriv = Priv;
        s->listen_subns_listhead=NULL;

        if(NULL == pstSubNs)
        {
            s->status = SOMB_SERVICE_STATUS_INVALID;
        }
        else
        {
            s->status = SOMB_SERVICE_STATUS_READY;
        }

        HASH_ADD_INT(SubNsServieList,ServiceId , s);  /* id is the key field */
        SOMB_LOG_INFO("somb mng:(%s) sub ns(%d) add new ServiceId(%08x) ServiceType(%d) ",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,ServiceId,ServiceType);
    }
    else
    {
        SOMB_LOG_INFO("somb mng:(%s) sub ns(%d) modify ServiceId(%08x) ServiceType(%d) Priv(%p)",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,ServiceId,ServiceType,Priv);
        //s->ServiceId = ServiceId;
        //s->ServiceType = ServiceType;

        if(Priv)
            s->ServicePriv = Priv;
    }

    if(NULL != pstSubNs)
        s->pstSubNs = pstSubNs;

    return s;
}

struct somb_mngr_sub_ns *pstSOMB_Manager_SaveSubNs(uint32_t sub_ns_id,char *sub_ns_addr,char* sub_ns_name)
{
    if((NULL == sub_ns_addr))
    {
        SOMB_LOG_ERR("somb mng:(%s) sub ns(%d) add new subns(%d) addr(%s) failed,param wrong",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,sub_ns_id,sub_ns_addr);
        return NULL;
    }

    struct somb_mngr_sub_ns *tmp=NULL;
    tmp = (struct somb_mngr_sub_ns*)malloc(sizeof *tmp);
    if(NULL == tmp)
    {
        SOMB_LOG_ERR("somb mng:(%s) sub ns(%d) add new subns(%d) addr(%s) malloc failed",pstLocalSubNs->localexename,
                pstLocalSubNs->sub_ns_id,sub_ns_id,sub_ns_addr);
        return NULL;
    }

    memset(tmp,0,sizeof(struct somb_mngr_sub_ns));
    somb_spinlock_acquire(&SubNsListLock);
    tmp->sub_ns_id = sub_ns_id;
    sprintf(tmp->sub_ns_addr,"%s",sub_ns_addr);
    sprintf(tmp->sub_ns_name,"%s",(NULL==sub_ns_name)?"NULL":sub_ns_name);

    pthread_condattr_t astCondAttr;
    pthread_condattr_setclock(&astCondAttr , CLOCK_MONOTONIC);
    pthread_cond_init(&(tmp->cond) , &astCondAttr);
    pthread_mutex_init(&tmp->mutex_lock,NULL);
    pthread_condattr_destroy(&astCondAttr);

    if(sub_ns_id != pstLocalSubNs->sub_ns_id)
    {
        tmp->skfd = socket(AF_UNIX,SOCK_STREAM|SOCK_CLOEXEC,0);
        struct sockaddr_un dest_addr_un;
        memset(&dest_addr_un,0,sizeof(struct sockaddr_un));
        dest_addr_un.sun_family =  AF_UNIX;
        sprintf(dest_addr_un.sun_path,"%s",tmp->sub_ns_addr);

        struct sockaddr_un src_addr_un;
        memset(&src_addr_un,0,sizeof(struct sockaddr_un));
        src_addr_un.sun_family =  AF_UNIX;
        sprintf(src_addr_un.sun_path,SOMB_SUB_NS_ADDR_FMT2,pstLocalSubNs->sub_ns_id,sub_ns_id);
        unlink(src_addr_un.sun_path);
        if(bind(tmp->skfd,(struct sockaddr*)&src_addr_un,SUN_LEN(&src_addr_un)) < 0)
        {
            SOMB_LOG_ERR("somb mng:(%s) pid(%d) subns_id(%d) bind addr for subns_id(%d) with srcaddr(%s) failed(%s)",pstLocalSubNs->localexename, pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,tmp->sub_ns_id,src_addr_un.sun_path,strerror(errno));
            close(tmp->skfd);
            free(tmp);

            /*try to find the exist subns*/
            tmp = NULL;
            HASH_FIND_INT(SubNsList,&(sub_ns_id),tmp) ;
            if(NULL == tmp)
            {
                goto finish;
            }
            else
            {
                SOMB_LOG_WARN("somb mng:(%s) pid(%d) subns_id(%d) save subns_id(%d) with srcaddr(%s) already success",pstLocalSubNs->localexename, pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,tmp->sub_ns_id,src_addr_un.sun_path);
                goto finish;
            }
        }
        else
        {
            SOMB_LOG_INFO("somb mng:(%s) pid(%d) subns_id(%d) bind addr for subns_id(%d) with srcaddr(%s) success",pstLocalSubNs->localexename, pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,tmp->sub_ns_id,src_addr_un.sun_path);

        }

        if(connect(tmp->skfd,(struct sockaddr*)&dest_addr_un,SUN_LEN(&dest_addr_un)) < 0)
        {
            tmp->SubNsStatus = SOMB_SUBNS_STATUS_OFFLINE;
            SOMB_LOG_ERR("somb mng:(%s) pid(%d) subns_id(%d) connect to new subns_id(%d) (%s) failed",pstLocalSubNs->localexename, pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,tmp->sub_ns_id,tmp->sub_ns_addr);
            close(tmp->skfd);
            free(tmp);
            goto finish;
        }

    }

    tmp->SubNsStatus = SOMB_SUBNS_STATUS_ONLINE;
    HASH_ADD_INT(SubNsList, sub_ns_id, tmp);  /* id is the key field */
    SOMB_LOG_INFO("somb mng:(%s) sub ns(%d) add new subns(%d) ,addr(%s) success,skfd=%d",pstLocalSubNs->localexename,
        pstLocalSubNs->sub_ns_id,sub_ns_id,sub_ns_addr,tmp->skfd);

finish:
    somb_spinlock_release(&SubNsListLock);
    return tmp;
}

ST_SOMB_LIB_SERVICE_LIST* pstSOMBManagerCheckServiceInfo(uint32_t ServiceId)
{
    ST_SOMB_LIB_SERVICE_LIST *s;
    HASH_FIND_INT(SubNsServieList, &ServiceId, s);/* s: output pointer */
    return s;
}

ST_SOMB_LIB_SERVICE_LIST* pstSOMBManagerPureGetServiceInfo(uint32_t ServiceId)
{
    ST_SOMB_LIB_SERVICE_LIST *s;

    pthread_mutex_lock(&SubsNsMutexLock);
    HASH_FIND_INT(SubNsServieList, &ServiceId, s);/* s: output pointer */
    pthread_mutex_unlock(&SubsNsMutexLock);
    return s;

}
ST_SOMB_LIB_SERVICE_LIST* pstSOMBManagerGetServiceInfo(uint32_t ServiceId)
{
    //uint32_t loop  = 0;
    int32_t ret;
    ST_SOMB_LIB_SERVICE_LIST *s;

    pthread_mutex_lock(&SubsNsMutexLock);
    HASH_FIND_INT(SubNsServieList, &ServiceId, s);/* s: output pointer */
    if((NULL == s)||((s->ServiceType == SOMB_SVC_TYPE_METHOD)&&(s->status == SOMB_SERVICE_STATUS_INVALID)))
    {

        somb_msg_get_service_req msgreq = somb_msg_get_service_req_init_zero;

        msgreq.service_id = ServiceId;
        msgreq.sub_ns_id = pstLocalSubNs->sub_ns_id;
        msgreq.pid = pstLocalSubNs->localpid;
        msgreq.has_subns = true;
        msgreq.subns.sub_ns_id = pstLocalSubNs->sub_ns_id;

        sprintf(msgreq.subns.sub_ns_addr,"%s",pstLocalSubNs->sub_ns_addr);
        //get service from daemon
        pthread_mutex_lock(&(pstLocalSubNs->sk_lock));

        ret = i32SOMB_Msg_Send_2_Daemon(somb_daemon_msgid_get_service,\
            &msgreq,somb_msg_get_service_req_fields,somb_msg_get_service_req_size);
        if(SOMB_OK != ret)
        {
            SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) get ServiceId(%08x) failed!",pstLocalSubNs->localexename,
                    pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,ServiceId);
            pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
            goto fail;
        }
        
        ret = i32SOMB_Manager_Poll(somb_daemon_msgid_get_service,1500);
        if(ret != SOMB_OK)
        {
            SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) get ServiceId(%08x) wait response failed!",pstLocalSubNs->localexename, pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,ServiceId);
            pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
            goto fail;
        }
        somb_msg_get_service_resp *msgresp = (somb_msg_get_service_resp *)(pstLocalSubNs->cond_MsgData);
        if(!msgresp->has_subns)
        {
            SOMB_LOG_ERR("somb mng: (%s)sub ns(%d) get serviceid(%08x) response,but no sub ns",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,msgresp->service_id);
            free(pstLocalSubNs->cond_MsgData);
            pstLocalSubNs->cond_MsgData =NULL;
            pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
            goto fail;
        }

        struct somb_mngr_sub_ns *pstSubNs = pstSOMB_Manager_Find_SubNs(msgresp->subns.sub_ns_id);
        if(NULL ==  pstSubNs)
        {
            //save sub ns
            pstSubNs = pstSOMB_Manager_SaveSubNs(msgresp->subns.sub_ns_id,msgresp->subns.sub_ns_addr,msgresp->subns.sub_ns_name);
        }
#if 0
        s = pstSOMB_Manager_SaveService(msgresp->service_id,msgresp->service_type,NULL, \
                                msgresp->subns.sub_ns_id,msgresp->subns.sub_ns_addr,msgresp->subns.sub_ns_name);
#endif

        s = pstSOMB_Manager_SaveService(msgresp->service_id,msgresp->service_type,NULL, pstSubNs);
        if(NULL == s)
        {
            SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) get ServiceId(%08x) but save failed!",pstLocalSubNs->localexename, pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,ServiceId);
            free(pstLocalSubNs->cond_MsgData);
            pstLocalSubNs->cond_MsgData =NULL;
            pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
            goto fail;
        }
        free(pstLocalSubNs->cond_MsgData);
        pstLocalSubNs->cond_MsgData=NULL;

        pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));

        SOMB_LOG_DEBUG("somb mng: (%s)sub ns(%d) get serviceid(%08x) from daemon success",pstLocalSubNs->localexename,pstLocalSubNs->sub_ns_id,ServiceId);
    }
    
    pthread_mutex_unlock(&SubsNsMutexLock);
    return s;

fail:
    pthread_mutex_unlock(&SubsNsMutexLock);
    return NULL;

}

int32_t i32SOMB_Manager_RegisterListenEvent_2_Daemon(struct somb_mngr_reg_serviceitem pui32ServiceList[],uint32_t ServiceSize,bool reg)
{
    int32_t ret;
    int loopMax = (ServiceSize%256 == 0) ? (ServiceSize/256) : (ServiceSize/256 + 1);

    for(int loop1 = 0; loop1 < loopMax ; loop1++)
    {
        uint32_t base_index = loop1 * 256;

        somb_event_listener_reg_2_bus listener_msg= somb_event_listener_reg_2_bus_init_zero;

        uint32_t loopMaxSize = (loopMax <= 1)? ServiceSize : ((loop1 == loopMax-1)? ServiceSize%256 : 256);
        listener_msg.service_item_count = loopMaxSize;
        listener_msg.flags_count = loopMaxSize;
        listener_msg.reg = reg;
        uint32_t loop=0;
        for(; loop < loopMaxSize; loop++)
        {
            listener_msg.service_item[loop].service_id = pui32ServiceList[base_index + loop].ServiceId;
            listener_msg.service_item[loop].service_type = pui32ServiceList[base_index + loop].ServiceType;
            if(listener_msg.service_item[loop].service_type == SOMB_SVC_TYPE_EVENT)
            {
                listener_msg.flags[loop]  = pui32ServiceList[base_index + loop].ServiceFlags;
            }
            else
            {
                listener_msg.flags[loop]  = 0;
            }
        }

        listener_msg.has_subns = true;
        snprintf(listener_msg.subns.sub_ns_name, 32, "%s",pstLocalSubNs->localexename);
        listener_msg.subns.sub_ns_id = pstLocalSubNs->sub_ns_id;
        sprintf(listener_msg.subns.sub_ns_addr,"%s",pstLocalSubNs->sub_ns_addr1);

        
        pthread_mutex_lock(&(pstLocalSubNs->sk_lock));

        ret = i32SOMB_Msg_Send_2_Daemon(somb_daemon_msgid_listen_event_2_bus,\
            &listener_msg,somb_event_listener_reg_2_bus_fields,somb_event_listener_reg_2_bus_size);
        if(SOMB_OK != ret)
        {
            SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) reg(%d) event listener to bus  failed!",pstLocalSubNs->localexename, pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id,reg);
            pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
            return SOMB_FAIL;
        }

        pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
    }
    return ret;
}

int32_t i32SOMB_Manager_ChangeServiceStatusWithMask_2_Daemon(somb_msg_set_service_available_with_mask *MaskAvailabeMsg)
{
    int32_t ret = SOMB_FAIL;

    pthread_mutex_lock(&(pstLocalSubNs->sk_lock));

    ret = i32SOMB_Msg_Send_2_Daemon(somb_daemon_msgid_service_change_available_with_mask,\
        MaskAvailabeMsg,somb_msg_set_service_available_with_mask_fields, somb_msg_set_service_available_with_mask_size);
    if(SOMB_OK != ret)
    {
        SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) change service available status with mask to bus  failed!",\
                     pstLocalSubNs->localexename, \
                     pstLocalSubNs->localpid, \
                     pstLocalSubNs->sub_ns_id);
    }

    pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));

    return ret;
}

int32_t i32SOMB_Manager_ChangeServiceStatus_2_Daemon(somb_msg_set_service_available *ServiceAvailableMsg)
{
    int32_t ret = SOMB_FAIL;

    
    pthread_mutex_lock(&(pstLocalSubNs->sk_lock));

    ret = i32SOMB_Msg_Send_2_Daemon(somb_daemon_msgid_service_change_available,\
        ServiceAvailableMsg,somb_msg_set_service_available_fields, somb_msg_set_service_available_size);
    if(SOMB_OK != ret)
    {
        SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) change service available status to bus  failed!",\
                     pstLocalSubNs->localexename, \
                     pstLocalSubNs->localpid, \
                     pstLocalSubNs->sub_ns_id);
    }

    pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));

    return ret;
}
int32_t i32SOMB_Manager_RegisterService_Daemon(struct somb_mngr_reg_serviceitem pui32ServiceList[],uint32_t Size)
{
    int32_t ret = SOMB_OK;


    int32_t regnum = 0;
    do{
        somb_msg_reg regmsg = somb_msg_reg_init_zero;

        uint32_t loop = 0;

        uint32_t loopMaxSize = (Size - regnum)>=256?256:Size - regnum;
        for(;loop<loopMaxSize;loop++)
        {
            regmsg.service_item[loop].service_id = pui32ServiceList[regnum].ServiceId;
            regmsg.service_item[loop].service_type = pui32ServiceList[regnum].ServiceType;
            regnum ++;
        }

        regmsg.service_item_count = loop;
        regmsg.has_subns = true;
        regmsg.subns.sub_ns_id = pstLocalSubNs->sub_ns_id;
        snprintf(regmsg.subns.sub_ns_addr,sizeof(regmsg.subns.sub_ns_addr),"%s",pstLocalSubNs->sub_ns_addr1);

        pthread_mutex_lock(&(pstLocalSubNs->sk_lock));

        ret = i32SOMB_Msg_Send_2_Daemon(somb_daemon_msgid_register,\
            &regmsg,somb_msg_reg_fields,somb_msg_reg_size);
        if(SOMB_OK != ret)
        {
            SOMB_LOG_ERR("somb mng:(%s) pid(%d) sub ns id(%d) reg service to bus  failed!",pstLocalSubNs->localexename,
                    pstLocalSubNs->localpid,pstLocalSubNs->sub_ns_id);
            pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
            return SOMB_FAIL;
        }

        pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));


    }while(regnum < Size);

    return ret;
}



void vSOMB_Manager_Save_Method_AsyncCB(uint32_t ServiceId, PF_SOMB_METHOD_ASYNC_CB pfMthAsyncFunc)
{
    ST_SOMB_LIB_SERVICE_LIST * pstLibServiceList = pstSOMBManagerGetServiceInfo(ServiceId);
    if(NULL == pstLibServiceList)
    {
        return ;
    }

    pstLibServiceList->AsyncCB = (void *)pfMthAsyncFunc;
}

PF_SOMB_METHOD_ASYNC_CB pfSOMB_Manager_Get_Method_AsyncCB(uint32_t ServiceId)
{

    ST_SOMB_LIB_SERVICE_LIST * pstLibServiceList = pstSOMBManagerGetServiceInfo(ServiceId);
    if(NULL == pstLibServiceList)
    {
        return NULL;
    }

    return (PF_SOMB_METHOD_ASYNC_CB)pstLibServiceList->AsyncCB;
}

void vSOMB_Manager_Report_Service_OnAvailable(uint8_t *buffer,uint32_t buffer_len)
{
    somb_service_available_msg msg = somb_service_available_msg_init_zero;

    pb_istream_t stream = pb_istream_from_buffer(buffer,buffer_len);
    if(false == pb_decode(&stream, somb_service_available_msg_fields, &msg))
    {
        SOMB_LOG_ERR("somb mng: (%s)sub ns(%d)  service on available msg decode failed,%s",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,PB_GET_ERROR(&stream));
        return;
    }

    uint32_t ServiceId = msg.service_id;
    ST_SOMB_LIB_SERVICE_LIST *s;
    HASH_FIND_INT(SubNsServieList, &ServiceId, s);/* s: output pointer */
    if(NULL != s)
    {
        SOMB_LOG_DEBUG("somb mng: service(%08x) type=%d provided by(%s %d) available(%d) has_subns(%d) priv=%p",
                      ServiceId,msg.service_type,msg.subns.sub_ns_name,msg.subns.sub_ns_id,msg.available,msg.has_subns,s->ServicePriv);

        //if receive service on available , the service should be checked and modified if necessary         
        if(s->ServiceType != msg.service_type)
        {
            SOMB_LOG_ERR("somb mng: (%s)sub ns(%d)  service(%08x) type mismatch(Listen:%d,Deploy:%d)",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,ServiceId,s->ServiceType,msg.service_type);
            return ;
        }

        if(msg.available)
        {
            s->status = SOMB_SERVICE_STATUS_READY;
        }
        else
        {
            s->status = SOMB_SERVICE_STATUS_OFFLINE;
        }

        do{
            if(false == msg.has_subns)
            {
                //keep the subns info
                break;
            }

            //update the service's subns info
            struct somb_mngr_sub_ns *tmp;
            somb_sub_ns *tmp_sub_ns = &(msg.subns);
            somb_spinlock_acquire(&SubNsListLock);
            HASH_FIND_INT(SubNsList,&(tmp_sub_ns->sub_ns_id),tmp) ;
            if(tmp == NULL)
            {
                tmp = (struct somb_mngr_sub_ns*)malloc(sizeof *tmp);
                if(NULL == tmp)
                {
                    SOMB_LOG_ERR("somb mng: (%s)sub ns(%d) malloc for new subns(%d) failed",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,tmp_sub_ns->sub_ns_id);
                    return ;
                }
                memset(tmp,0,sizeof(struct somb_mngr_sub_ns));
                tmp->sub_ns_id = tmp_sub_ns->sub_ns_id;
                tmp->resp_data_head = NULL;

                pthread_condattr_t astCondAttr;
                pthread_condattr_setclock(&astCondAttr , CLOCK_MONOTONIC);
                pthread_cond_init(&(tmp->cond) , &astCondAttr);

                pthread_mutex_init(&tmp->mutex_lock,NULL);

                sprintf(tmp->sub_ns_addr,"%s",tmp_sub_ns->sub_ns_addr);
                sprintf(tmp->sub_ns_name,"%s",tmp_sub_ns->sub_ns_name);

                HASH_ADD_INT(SubNsList, sub_ns_id, tmp);  /* id is the key field */

                SOMB_LOG_INFO("somb mng: (%s)sub ns(%d) new sub ns(%d) added",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,tmp_sub_ns->sub_ns_id);

                pthread_condattr_destroy(&astCondAttr);

                somb_manager_connect_remote_subns(tmp);
                s->pstSubNs = tmp;
            }
            else
            {
                s->pstSubNs = tmp;
            }
            somb_spinlock_release(&SubNsListLock);
        }while(0);

    }
    else
    {
        SOMB_LOG_ERR("somb mng: (%s)sub ns(%d)  service(%08x) on available  s=%p",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,ServiceId,s);
        return;
    }

    if((NULL != s)&&(SOMB_SVC_TYPE_METHOD == s->ServiceType)&&(NULL != s->ServicePriv))
    {
        PF_SOMB_METHOD_ON_AVAILABLE_CB pfFunc = s->ServicePriv;
        pfFunc(ServiceId,msg.available);
    }
}

void vSOMB_Manager_Resp_Process(uint32_t MsgId,const pb_msgdesc_t * fields,uint32_t MsgSize,uint8_t *buffer,uint32_t buffer_len)
{

    pthread_mutex_lock(&(pstLocalSubNs->sk_lock));
    if(NULL != pstLocalSubNs->cond_MsgData)
    {
        SOMB_LOG_ERR("somb mng: (%s)sub ns(%d) get MsgId(%d) resp but msgdata is %p,msgid=%d",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,MsgId,pstLocalSubNs->cond_MsgData,pstLocalSubNs->cond_MsgId);
        free(pstLocalSubNs->cond_MsgData);
    }

    //malloc memory to store the response data
    pstLocalSubNs->cond_MsgData = malloc(MsgSize);
    if(NULL == pstLocalSubNs->cond_MsgData)
    {
        SOMB_LOG_ERR("somb mng: (%s)sub ns(%d) get MsgId(%d) resp but malloc for msgdata(%d) failed",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,MsgId,MsgSize);
        goto fail;
    }

    memset(pstLocalSubNs->cond_MsgData,0,MsgSize);
        
    pb_istream_t stream = pb_istream_from_buffer(buffer,buffer_len);
    if(false == pb_decode(&stream, fields, pstLocalSubNs->cond_MsgData))
    {
        SOMB_LOG_ERR("somb mng: (%s)sub ns(%d) get MsgId(%d) resp decode failed,%s",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,MsgId,PB_GET_ERROR(&stream));
        goto fail1;
    }
    pstLocalSubNs->cond_MsgId =  MsgId;
    pthread_cond_broadcast(&(pstLocalSubNs->sk_cond));
    pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
    return;
fail1:
    free(pstLocalSubNs->cond_MsgData);
    pstLocalSubNs->cond_MsgData=NULL;
fail:
    pthread_mutex_unlock(&(pstLocalSubNs->sk_lock));
    /*notify requester */
}

void vSOMB_Manager_Recv_Listen_SubNs_Process(uint8_t *buffer,uint32_t buffer_len)
{

    somb_event_listener_reg_2_subns event_listen_msg = somb_event_listener_reg_2_subns_init_zero;
    pb_istream_t stream = pb_istream_from_buffer(buffer,buffer_len);
    if(false == pb_decode(&stream, somb_event_listener_reg_2_subns_fields,&event_listen_msg))
    {
        SOMB_LOG_ERR("somb mng: (%s)sub ns(%d) event listen msg decode failed,%s",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,PB_GET_ERROR(&stream));
        return ;
    }

    ST_SOMB_LIB_SERVICE_LIST *s;
    HASH_FIND_INT(SubNsServieList, &(event_listen_msg.service_id), s);/* s: output pointer */
    if(NULL == s)
    {
        SOMB_LOG_ERR("somb mng: (%s)sub ns(%d) event listen msg ,but ServiceId(%08x) not in manager servicelist",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,event_listen_msg.service_id);
        return ;
    }

    uint32_t loop = 0;
    for(;loop < event_listen_msg.subns_count;loop++)
    {
        struct somb_mngr_sub_ns *tmp;
        somb_sub_ns *tmp_sub_ns = &(event_listen_msg.subns[loop]);
        somb_spinlock_acquire(&SubNsListLock);
        HASH_FIND_INT(SubNsList,&(tmp_sub_ns->sub_ns_id),tmp) ;
        if(tmp == NULL)
        {
            tmp = (struct somb_mngr_sub_ns*)malloc(sizeof *tmp);
            if(NULL == tmp)
            {
                SOMB_LOG_ERR("somb mng: (%s)sub ns(%d) event listen msg ,malloc for new subns(%d) failed",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,tmp_sub_ns->sub_ns_id);
                return ;
            }
            memset(tmp,0,sizeof(struct somb_mngr_sub_ns));
            tmp->sub_ns_id = tmp_sub_ns->sub_ns_id;
            tmp->resp_data_head = NULL;

            pthread_condattr_t astCondAttr;
            pthread_condattr_setclock(&astCondAttr , CLOCK_MONOTONIC);
            pthread_cond_init(&(tmp->cond) , &astCondAttr);

            pthread_mutex_init(&tmp->mutex_lock,NULL);

            snprintf(tmp->sub_ns_addr,sizeof(tmp->sub_ns_addr),"%s",tmp_sub_ns->sub_ns_addr);
            snprintf(tmp->sub_ns_name,sizeof(tmp->sub_ns_name),"%s",tmp_sub_ns->sub_ns_name);

            HASH_ADD_INT(SubNsList, sub_ns_id, tmp);  /* id is the key field */

            pthread_condattr_destroy(&astCondAttr);

            if(SOMB_OK != somb_manager_connect_remote_subns(tmp))
            {
                tmp->SubNsStatus = SOMB_SUBNS_STATUS_INVALID;
                SOMB_LOG_WARN("somb mng: (%s)sub ns(%d) add new sub ns(%d) wich is invalid now",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,tmp_sub_ns->sub_ns_id);
            }
            else
            {
                tmp->SubNsStatus = SOMB_SUBNS_STATUS_ONLINE;
                SOMB_LOG_INFO("somb mng: (%s)sub ns(%d) add new sub ns(%d) ",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,tmp_sub_ns->sub_ns_id);

            }
        }
        somb_spinlock_release(&SubNsListLock);

        if(event_listen_msg.reg)
        {
            bool find_in_listenlist = false;
            struct somb_mngr_service_subns_item *item11,*tmp11;
            DL_FOREACH_SAFE(s->listen_subns_listhead,item11,tmp11)
            {
                if(item11->pstSubNs->sub_ns_id == tmp->sub_ns_id)
                {
                    find_in_listenlist =true; 
                    SOMB_LOG_WARN("somb mng: (%s)sub ns(%d) event listen msg , new subns(%d) already in (0x%08x)'s listen list",\
                                  pstLocalSubNs->localexename, 
                                  pstLocalSubNs->sub_ns_id,
                                  tmp_sub_ns->sub_ns_id,
                                  s->ServiceId);
                    break;
                }
            }

            if(!find_in_listenlist)
            {
                struct somb_mngr_service_subns_item *subns_item = malloc(sizeof(struct somb_mngr_service_subns_item));
                if(NULL == subns_item)
                {
                    SOMB_LOG_ERR("somb mng: (%s)sub ns(%d) event listen msg ,malloc for new subns(%d) item failed",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,tmp_sub_ns->sub_ns_id);
                    return ;
                }

                subns_item->pstSubNs = tmp;

                DL_PREPEND(s->listen_subns_listhead,subns_item);
                SOMB_LOG_INFO("somb mng: (%s)sub ns(%d) recv reg service(%08x)'s event listen from subns(%d) success",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,event_listen_msg.service_id,tmp_sub_ns->sub_ns_id);

                item11 = subns_item;

            }

            //if the service is provided by owner;
            //get the event service id callback ,and get the event data through calling it 
            if((NULL != s->ServicePriv) && (event_listen_msg.flags[loop] & SOMB_SVC_FLAGS_EVENT_LISTEN_SYNC_DATA))
            {
                PF_SOMB_EVENT_ON_LISTEN_CB ev_on_listen_callback= (PF_SOMB_EVENT_ON_LISTEN_CB)(s->ServicePriv);
                void *event_data = NULL;
                uint32_t event_data_len=0;
                if(SOMB_OK == ev_on_listen_callback(s->ServiceId, &event_data, &event_data_len))
                {
                    int32_t i32SOMB_Event_Trigger_To_Subns(ST_SOMB_LIB_SERVICE_LIST *Service,void *EventData,uint32_t EventDataLen, struct somb_mngr_service_subns_item *dest_subns);
                    if(SOMB_OK != i32SOMB_Event_Trigger_To_Subns(s, event_data,event_data_len,item11))
                    {
                        SOMB_LOG_ERR("somb mng: (%s)sub ns(%d) trigger service(%08x)'s event to subns(%d) fail",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,s->ServiceId,item11->pstSubNs->sub_ns_id);
                    }

                    if(NULL != event_data)
                        free(event_data); 
                }
            }

        }
        else
        {
            struct somb_mngr_service_subns_item *item1,*tmp1;
            DL_FOREACH_SAFE(s->listen_subns_listhead,item1,tmp1)
            {
                if(item1->pstSubNs->sub_ns_id == tmp_sub_ns->sub_ns_id)
                {
                    SOMB_LOG_INFO("somb mng: (%s)sub ns(%d) recv unreg service(%08x)'s event listen from subns(%d) success",pstLocalSubNs->localexename, pstLocalSubNs->sub_ns_id,event_listen_msg.service_id,tmp_sub_ns->sub_ns_id);
                    DL_DELETE(s->listen_subns_listhead,item1);
                    free(item1);
                    break;
                }
            }

        }
    }
}

void vsomb_manager_get_service_statistic(somb_subns_work_status *statistic)
{
    if(NULL == statistic)
        return;
    ST_SOMB_LIB_SERVICE_LIST *s=NULL,*tmp=NULL;
    int service_statistic_count =0;
    HASH_ITER(hh,SubNsServieList ,s,tmp)
    {
        if(service_statistic_count >=128)
        {
            break;
        }

        statistic->service_statistic[service_statistic_count].service_id = s->ServiceId;
        statistic->service_statistic[service_statistic_count].service_type = s->ServiceType;
        if((s->pstSubNs)&&(s->pstSubNs->sub_ns_id == pstLocalSubNs->sub_ns_id))
        {
            statistic->service_statistic[service_statistic_count].service_role = 0;
        }
        else
        {
            statistic->service_statistic[service_statistic_count].service_role = 1;
        }
        statistic->service_statistic[service_statistic_count].Count= s->ServiceCount;
        statistic->service_statistic[service_statistic_count].FailCount= s->ServiceFailCount;

        service_statistic_count++;
    }

    statistic->service_statistic_count = service_statistic_count;
}
